package com.dc.rxjava;

import android.util.Log;

import java.util.ArrayList;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.observables.GroupedObservable;

/**
 * 转变操作符
 *
 * @author 冰柠檬dc
 */
public class TransformOperatorUtils {

    public static final String TAG = "RxJava";

    public static ArrayList<Person> getPersonList() {
        // [Person{name='0', planList=[Plan{actionList=[0, 2]}, Plan{actionList=[0, 3]}]}, Person{name='1', planList=[Plan{actionList=[1, 4]}, Plan{actionList=[2, 6]}]}]
        ArrayList<Person> list = new ArrayList<>();
        for (int i = 0; i < 2; i++) {
            Person person = new Person(String.valueOf(i));
            ArrayList<Plan> planList = new ArrayList<>();
            for (int j = 1; j < 3; j++) {
                ArrayList<Integer> actions = new ArrayList<>();
                actions.add(i * j);
                actions.add((i + 1) * (j + 1));
                Plan plan = new Plan(actions);
                planList.add(plan);
            }
            person.setPlanList(planList);
            list.add(person);
        }
        return list;
    }

    /**
     * map操作符：可以将被观察者发送的数据类型转变成其他的类型
     *
     * @return Disposable
     */
    public static Disposable map() {
        // 现在有一个需求就是要将 Person 集合中的每个元素中的 Plan 的 action 打印出来。 首先用 map() 来实现
        // 可以看到 onNext() 用了嵌套 for 循环来实现，如果代码逻辑复杂起来的话，可能需要多重循环才可以实现。
        return Observable.fromIterable(getPersonList()).map(new Function<Person, ArrayList<Plan>>() {
            @Override
            public ArrayList<Plan> apply(Person person) throws Exception {
                return person.getPlanList();
            }
        }).subscribe(plans -> {
            for (Plan plan : plans) {
                ArrayList<Integer> actionList = plan.getActionList();
                for (Integer action : actionList) {
                    Log.e(TAG, "onNext:" + action);
                }
            }
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
    }

    /**
     * flatMap操作符：可以将被观察者发送的数据类型转变成其他的类型
     *
     * @return Disposable
     */
    public static Disposable flatMap() {
        // 现在有一个需求就是要将 Person 集合中的每个元素中的 Plan 的 action 打印出来。 首先用 map() 来实现
        // 可以看到 onNext() 用了嵌套 for 循环来实现，如果代码逻辑复杂起来的话，可能需要多重循环才可以实现。
        return Observable.fromIterable(getPersonList()).flatMap(new Function<Person, ObservableSource<Plan>>() {
            @Override
            public ObservableSource<Plan> apply(Person person) throws Exception {
                // TODO 这个if用于构造区分flatMap和concatMap，用flatMap会先执行name为1的数据，2s后再执行name为0的数据
                if ("0".equals(person.getName())) {
                    return Observable.fromIterable(person.getPlanList()).delay(2, TimeUnit.SECONDS);
                }
                return Observable.fromIterable(person.getPlanList());
            }
        }).flatMap(new Function<Plan, ObservableSource<Integer>>() {
            @Override
            public ObservableSource<Integer> apply(Plan plan) throws Exception {
                return Observable.fromIterable(plan.getActionList());
            }
        }).subscribe(action -> {
            Log.e(TAG, "onNext:" + action);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
    }

    /**
     * concatMap操作符：concatMap() 和 flatMap() 基本上是一样的，只不过 concatMap() 转发出来的事件是有序的，而 flatMap() 是无序的。
     *
     * @return Disposable
     */
    public static Disposable concatMap() {
        // 现在有一个需求就是要将 Person 集合中的每个元素中的 Plan 的 action 打印出来。 首先用 map() 来实现
        // 可以看到 onNext() 用了嵌套 for 循环来实现，如果代码逻辑复杂起来的话，可能需要多重循环才可以实现。
        return Observable.fromIterable(getPersonList()).concatMap(new Function<Person, ObservableSource<Plan>>() {
            @Override
            public ObservableSource<Plan> apply(Person person) throws Exception {
                // TODO 这个if用于构造区分flatMap和concatMap， concatMap要保证发送的顺序会等2s执行name为0的数据，然后再执行name为1的数据
                if ("0".equals(person.getName())) {
                    return Observable.fromIterable(person.getPlanList()).delay(2, TimeUnit.SECONDS);
                }
                return Observable.fromIterable(person.getPlanList());
            }
        }).concatMap(new Function<Plan, ObservableSource<Integer>>() {
            @Override
            public ObservableSource<Integer> apply(Plan plan) throws Exception {
                return Observable.fromIterable(plan.getActionList());
            }
        }).subscribe(action -> {
            Log.e(TAG, "onNext:" + action);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
    }

    /**
     * buffer操作符：从需要发送的事件当中获取一定数量的事件，并将这些事件放到缓冲区当中一并发出
     *
     * @return Disposable
     */
    public static Disposable buffer() {
        return Observable.just(1, 2, 3, 4, 5).buffer(2, 1).subscribe(list -> {
            Log.e(TAG, "缓冲区大小:" + list.size());
            for (Integer i : list) {
                Log.e(TAG, "onNext: " + i);
            }
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：(skip默认为count，这里指定为1，每次发送事件，指针都会往后移动一个元素再取值，直到指针移动到没有元素的时候就会停止取值。)
        // onSubscribe
        // 缓冲区大小： 2
        // onNext:1
        // onNext:2
        // 缓冲区大小： 2
        // onNext:2
        // onNext:3
        // 缓冲区大小： 2
        // onNext:3
        // onNext:4
        // 缓冲区大小： 2
        // onNext:4
        // onNext:5
        // 缓冲区大小： 1
        // onNext:5
        // onComplete
    }

    /**
     * groupBy操作符：将发送的数据进行分组，每个分组都会返回一个被观察者。
     *
     * @return Disposable
     */
    public static Disposable groupBy() {
        return Observable.just(5, 2, 3, 4, 1, 6, 8, 9, 7, 10).groupBy(new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer integer) throws Exception {
                return integer % 3;
            }
        }).subscribe(new Consumer<GroupedObservable<Integer, Integer>>() {
            @Override
            public void accept(GroupedObservable<Integer, Integer> integerIntegerGroupedObservable) throws Exception {
                Log.e(TAG, "onNext");
                Disposable subscribe = integerIntegerGroupedObservable.subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "GroupedObservable onNext: " + integerIntegerGroupedObservable.getKey() + ":" + integer);
                    }
                }, throwable -> {
                    Log.e(TAG, "GroupedObservable onError");
                }, () -> {
                    Log.e(TAG, "GroupedObservable onComplete");
                }, onSubscribe -> {
                    Log.e(TAG, "GroupedObservable onSubscribe");
                });
            }
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果： groupBy() 方法返回的参数是分组的名字，每返回一个值，那就代表会创建一个组，以上的代码就是将1~10的数据分成3组
        // onSubscribe
        // onNext
        // GroupedObservable onSubscribe
        // GroupedObservable onNext: 2:5
        // GroupedObservable onNext: 2:2
        // onNext
        // GroupedObservable onSubscribe
        // GroupedObservable onNext: 0:3
        // onNext
        // GroupedObservable onSubscribe
        // GroupedObservable onNext: 1:4
        // GroupedObservable onNext: 1:1
        // GroupedObservable onNext: 0:6
        // GroupedObservable onNext: 2:8
        // GroupedObservable onNext: 0:9
        // GroupedObservable onNext: 1:7
        // GroupedObservable onNext: 1:10
        // GroupedObservable onComplete
        // GroupedObservable onComplete
        // GroupedObservable onComplete
        // onComplete
    }

    /**
     * scan操作符：将数据以一定的逻辑聚合起来。
     *
     * @return Disposable
     */
    public static Disposable scan() {
        return Observable.just(1, 3, 5, 7).scan(new BiFunction<Integer, Integer, Integer>() {
            @Override
            public Integer apply(Integer sum, Integer integer) throws Exception {
                Log.e(TAG, "sum:" + sum);
                Log.e(TAG, "integer:" + integer);
                return sum + integer;
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：当执行原Observable输入的第一个时不会回调call函数，也就是说第一次并不参与运算,直接输给订阅者，onNext的值会在下次发送时赋值给第一个参数sum
        // onSubscribe
        // onNext:1
        // sum:1
        // integer:3
        // onNext:4
        // sum:4
        // integer:5
        // onNext:9
        // sum:9
        // integer:7
        // onNext:16
        // onComplete
    }

    /**
     * window操作符：发送指定数量的事件时，就将这些事件分为一组。window 中的 count 的参数就是代表指定的数量，例如将 count 指定为2，那么每发2个数据就会将这2个数据分成一组。
     *
     * @return Disposable
     */
    public static Disposable window() {
        return Observable.just(1, 2, 3, 4, 5).window(2).subscribe(integerObservable -> {
            Log.e(TAG, "onNext");
            Disposable disposable = integerObservable.subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    Log.e(TAG, "integerObservable onNext: " + integer);
                }
            }, throwable -> {
                Log.e(TAG, "integerObservable onError");
            }, () -> {
                Log.e(TAG, "integerObservable onComplete");
            }, onSubscribe -> {
                Log.e(TAG, "integerObservable onSubscribe");
            });
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：window() 将 1~5 的事件分成了3组。
        // onSubscribe
        // onNext
        // integerObservable onSubscribe
        // integerObservable onNext:1
        // integerObservable onNext:2
        // integerObservable onComplete
        // onNext
        // integerObservable onSubscribe
        // integerObservable onNext:3
        // integerObservable onNext:4
        // integerObservable onComplete
        // onNext
        // integerObservable onSubscribe
        // integerObservable onNext:5
        // integerObservable onComplete
        // onComplete
    }
}
